package cn.ilikes.tools.nosql.jpa.clients.indexs.index.lucenes.asytools;

import org.I0Itec.zkclient.ZkClient;

import org.slf4j.Logger;

import cn.ilikes.tools.logs.Loggerfactory;
import cn.ilikes.tools.nosql.jpa.SysConfigs;
import cn.ilikes.tools.nosql.jpa.clients.indexs.ClientIndexingRunTimeException;
import cn.ilikes.tools.nosql.jpa.clients.indexs.index.lucenes.AsyLuceneClientIndex;
import cn.ilikes.tools.nosql.jpa.configs.indexs.IndexConfigs;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class AsyZookeeprLockFactory implements ZookeeprLockFactory {

	private Logger logger = Loggerfactory.instance(AsyZookeeprLockFactory.class);

	private ConcurrentHashMap<String, Object> lockMaps = new ConcurrentHashMap<String, Object>();

	private ZkClient zkClient4subChild;

	private AsyZookeeprLockListener zookeeprLockListener;

	private IndexConfigs indexConfigs;

	private String lockNode;

	public AsyZookeeprLockFactory(IndexConfigs indexConfigs, AsyZookeeprLockListener zookeeprLockListener) {
		super();
		this.indexConfigs = indexConfigs;
		this.zookeeprLockListener = zookeeprLockListener;
	}

	@Override
	public void start() {
		Loggerfactory.info(logger, "start  Asy Zookeepr Lock Factory ");
		lockNode = indexConfigs.getProperties().getProperty(SysConfigs.index_zookeeper_lock_node);

		String zookeeperHost = indexConfigs.getProperties().getProperty(SysConfigs.index_zookeeper_host_address);

		ZkClient zkClient4subChild = new ZkClient(zookeeperHost);

		zkClient4subChild.subscribeChildChanges(lockNode, zookeeprLockListener);

		this.setZkClient4subChild(zkClient4subChild);

		if (!zkClient4subChild.exists(lockNode)) {
			try {
				zkClient4subChild.createPersistent(lockNode);
			} catch (Exception e) {
				throw new ClientIndexingRunTimeException("Error  ", e);
			}
		}

	}

	@Override
	public void stop() {
		ZkClient zkClient4subChild = getZkClient4subChild();
		zkClient4subChild.close();
	}

	private ZkClient getZkClient4subChild() {
		return zkClient4subChild;
	}

	private void setZkClient4subChild(ZkClient zkClient4subChild) {
		this.zkClient4subChild = zkClient4subChild;
	}

	@Override
	public void addLock(String lockId, String time) {
		try {
			String lock = LockUilts.addlock(lockNode, lockId, time);
			if (Loggerfactory.isdebug(logger))
				Loggerfactory.debug(logger, "add  Lock " + lock);
			ZkClient zkClient4subChild = getZkClient4subChild();
			zkClient4subChild.createEphemeral(lock);
			lockMaps.put(lock, lock);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	@Override
	public void unLock(String lockId, String localLockId) {
		if (!lockId.equals(lockNode))return;
		String lock = LockUilts.getlock(lockNode, localLockId);
		if (Loggerfactory.isdebug(logger)) Loggerfactory.debug(logger, "un Lock " + lockId + "   " + localLockId);
		ZkClient zkClient4subChild = getZkClient4subChild();
		zkClient4subChild.delete(lock);
		lockMaps.remove(lock);
	}

	@Override
	public void release() {
		for (Map.Entry<String, Object> xx : lockMaps.entrySet()) {
			zkClient4subChild.delete(xx.getKey());
		}
		lockMaps.clear();
	}

	@Override
	public boolean islocalLock(String parents) {
		return (parents.equals(lockNode));
	}

}
